package pku;

import java.io.*;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.io.FileNotFoundException;
import java.io.IOException;

/**
 * 这是一个简单的基于内存的实现，以方便选手理解题意；
 * 实际提交时，请维持包名和类名不变，把方法实现修改为自己的内容；
 */
public class DefaultQueueStoreImpl extends QueueStore {

    public ConcurrentHashMap<String, Queue> queueMap = new ConcurrentHashMap<String, Queue>();
    private AtomicLong position = new AtomicLong(0);
    private boolean firstPut = true;
    private boolean firstGet = true;
    private FileChannel channel;
    private FileChannel inchannel;
    private String path = "/";

    public synchronized void put(String queueName, byte[] message) {

        if(firstPut){
//            File dir = new File(path);
//            if(!dir.exists()){
//                dir.mkdirs();
//            }

            File file = new File("./data/data.dat");
//            File file=new File("D:\\data\\data.dat");
            if(!file.exists()){
                try {
                    file.createNewFile();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
//            try {
//                RandomAccessFile f = new RandomAccessFile(file, "rw");
//                this.channel = f.getChannel();
//                this.inchannel = f.getChannel();
//            } catch (FileNotFoundException e) {
//                e.printStackTrace();
//            }

            try {
                this.channel = new FileOutputStream(file).getChannel();
                this.inchannel = new FileInputStream(file).getChannel();
//                this.inchannel = this.channel;
            } catch (FileNotFoundException e) {
                e.printStackTrace();
            }
        }
        firstPut = false;

        Queue queue;
        try {
            if(queueMap.get(queueName)==null){
                queue = new Queue(this.channel,this.inchannel,position);
                queueMap.put(queueName,queue);
                queue.put(message);
            }else {
                queue = queueMap.get(queueName);
                queue.put(message);
            }
        }catch (Exception ex){
        }
    }

    public  Collection<byte[]> get(String queueName, long offset, long num) {
        synchronized (this){
            if (firstGet){
                try {
                    for (Queue q:queueMap.values()){
                        q.flush();
                    }
                }catch (Exception e){
                }
                this.firstGet = false;
            }
        }

        try {
            return queueMap.get(queueName).get(offset, num);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
}

class Queue {

    private ByteBuffer buffer;
    private static final int BUFFER_SIZE = 4*1024;      //缓冲区大小
    private FileChannel channel;                       //刷盘通道
    private FileChannel inchannel;                      //读取通道
    private AtomicLong position;                        //当前文件中可供写入的第一个位置
    private ArrayList<AtomicLong[]> index;               //该队列在文件中存储块的索引，前面是偏移位置，后面是块序号
    private AtomicInteger messagecounter;               //待存入的下一条消息编号
    private AtomicInteger firstmsgindexofblock;         //当前缓冲区中存的第一条消息序号
    private static final byte MSGLENLEN = 2;            //消息长度的存储长度

    public Queue(FileChannel channel, FileChannel inchannel, AtomicLong position) throws FileNotFoundException {

        this.channel = channel;
        this.inchannel = inchannel;
        this.position = position;
        this.buffer = ByteBuffer.allocateDirect(BUFFER_SIZE);
        this.index = new ArrayList<>();
        this.messagecounter = new AtomicInteger(0);
        this.firstmsgindexofblock = new AtomicInteger(0);
    }

    public void put(byte[] message) throws IOException {
        //缓冲区存不下消息则先落盘
        if (buffer.remaining() < message.length + MSGLENLEN) {
            flush();
        }
        buffer.putShort((short) message.length);
        buffer.put(message);
        this.messagecounter.incrementAndGet();
    }

    public synchronized Collection<byte[]> get(long offset, long num) throws IOException {
        Collection<byte[]> results = new ArrayList<byte[]>();
        //消息所在的存储块号
        int searchindex = this.index.size() - 1;
        for (int i = 0; i < this.index.size() - 1; i++) {
            if (this.index.get(i)[1].get() <= offset && this.index.get(i + 1)[1].get() > offset) {
                searchindex = i;
                break;
            }
        }
        buffer.clear();
        inchannel.read(this.buffer, this.index.get(searchindex)[0].get());
        buffer.flip();
        //跳过前面部分
        int currentindex = (int) this.index.get(searchindex)[1].get();
        while (currentindex < offset) {
            int msglen = buffer.getShort();
            buffer.position(buffer.position() + msglen);
            currentindex++;
        }
        //开始读取消息
        while (num > 0) {
            if (buffer.position()>=this.index.get(searchindex)[2].get()){
                break;
            }

            int msglen = buffer.getShort();
            byte[] msg = new byte[msglen];
            buffer.get(msg);
            results.add(msg);
            num--;
        }

        //消息存在多个存储块中
        while (num>0){
            searchindex++;
            if (searchindex > index.size() - 1) {
                break;
            }
            buffer.clear();
            inchannel.read(this.buffer, this.index.get(searchindex)[0].get());
            buffer.flip();
            while (num > 0) {
                if (buffer.position()>=this.index.get(searchindex)[2].get()){
                    break;
                }
                int msglen = buffer.getShort();
                byte[] msg = new byte[msglen];
                buffer.get(msg);
                results.add(msg);
                num--;
            }
        }
        return results;
    }

    public void flush() throws IOException {
        int bufferlen = buffer.position();
        buffer.flip();
        channel.write(buffer, this.position.get());
        buffer.clear();
        //一个缓存区对应一个存储块，刷盘时生成该存储块的索引
        AtomicLong[] indexnext = {new AtomicLong(position.longValue()), new AtomicLong(this.firstmsgindexofblock.get()),new AtomicLong(bufferlen)};
        this.index.add(indexnext);
        position.getAndAdd(bufferlen);
        firstmsgindexofblock.set(this.messagecounter.get());
    }

}


